package main.niit

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties

//本业务分析使用年限的车辆数
object vehicleAge {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.OFF)
    val spark =SparkSession.builder()
      .appName("vehicleAge")
      .master("local[*]")
      .getOrCreate()

    val jdbcDF =spark.read.format("jdbc")
      .option("url","jdbc:mysql://192.168.56.104:3306/cardata?useSSL=false")
      .option("driver","com.mysql.jdbc.Driver")
      .option("dbtable","data")
      .option("user","root")
      .option("password","root")
      .load()

    import spark.implicits._
    val prop =new Properties()
    prop.put("user","root")
    prop.put("password","root")
    prop.put("driver","com.mysql.jdbc.Driver")


    val yearExpr = split($"广告创作日期", "/")(0).cast("int") //使用"/"分割字符串，并取数组第一个元素，将其转换为整数类型。
    val filterDF=jdbcDF.select($"注册年份",yearExpr.alias("广告创作年份"))
      .filter($"注册年份"<=2016)
    val resultDF=filterDF.withColumn("使用年份",$"广告创作年份" - $"注册年份")  //添加新列"使用年份"="广告创作年份"-"注册年份"
      .withColumn("数量",lit(1))
      .groupBy("使用年份")
      .agg(
        sum("数量").alias("车辆数量")
      )
      .orderBy($"使用年份".asc)//升序排序
//      .show

    resultDF.write.mode("append")
      .jdbc("jdbc:mysql://192.168.56.104:3306/cardata?useSSL=false","cardata.vehicleAge",prop)
  }

}
